package prictice.transaction1.daily;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.transactional.TransactionalTopologyBuilder;
import prictice.transaction1.MyTxSpout;

/**
 * @Package prictice.transaction1
 * @Description:
 * @Author elwyn
 * @Date 2017/9/16 20:32
 * @Email elonyong@163.com
 */
public class MyDailyTopo {
    public static void main(String[] args) {
        TransactionalTopologyBuilder transactionalTopologyBuilder =
                new TransactionalTopologyBuilder("ttbId", "spoutid", new MyTxSpout(), 1);
        transactionalTopologyBuilder.setBolt("bolt1", new MyDailyBatchBolt(), 3).shuffleGrouping("spoutid");
        transactionalTopologyBuilder.setBolt("committer", new MyDailyCommitter(), 1).shuffleGrouping("bolt1");
        Config config = new Config();
        if (args.length > 0) {
            try {
                StormSubmitter.submitTopology(args[0], config, transactionalTopologyBuilder.buildTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("mytopology", config, transactionalTopologyBuilder.buildTopology());
        }
    }
}
